package cn.itcast.streaming.sink;

import cn.itcast.streaming.bean.CustomRuleAlarmResultModel;
import cn.itcast.streaming.utils.ConfigLoader;
import cn.itcast.streaming.utils.DateUtil;
import cn.itcast.streaming.utils.MongoUtil;
import com.google.common.collect.Lists;
import com.mongodb.BasicDBObject;
import com.mongodb.MongoClient;
import com.mongodb.client.MongoCollection;
import com.mongodb.client.MongoCursor;
import com.mongodb.client.MongoDatabase;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
import org.bson.Document;

import java.util.*;
import java.util.stream.Collectors;

//自定义告警规则数据的存储
public class CustomRuleAlarmMongoSink extends RichSinkFunction<ArrayList<CustomRuleAlarmResultModel>> {

    private MongoClient mongoClient = null;
    private MongoDatabase database;
    private MongoCollection<Document> collection;

    @Override
    public void open(Configuration parameters) throws Exception {
        super.open(parameters);
        mongoClient = MongoUtil.getConnect();
        //获取当前操作的数据库
        database = mongoClient.getDatabase(ConfigLoader.getProperty("mongo.source"));
        //获取表
        collection = database.getCollection("custom_rule_alarm");
    }

    @Override
    public void close() throws Exception {
        super.close();
        mongoClient.close();
    }

    @Override
    public void invoke(ArrayList<CustomRuleAlarmResultModel> value, Context context) throws Exception {
        //在某些情况下，某个车架id的某个窗口的数据可能同时满足多个监控类型，因此需要根据监控类型进行分组，同一个监控类型的数据放入一个组中
        Map<Integer, List<CustomRuleAlarmResultModel>> ruleAlarmModelListMap = value.stream()
                .collect(Collectors.groupingBy(ruleAlarmRuleModel -> ruleAlarmRuleModel.getMonitorId()));

        //获取到map的所有key
        Set<Integer> monitorIdSet = ruleAlarmModelListMap.keySet();
        Iterator<Integer> iterator = monitorIdSet.iterator();

        //循环所有监控id 全部车辆告警 车型告警 区域告警 大客户告警 车辆告警 车系告警 车辆在线时长告警
        while (iterator.hasNext()){
            Integer monitorId = iterator.next();
            LinkedList<CustomRuleAlarmResultModel> ruleAlarmResultModelList = Lists.newLinkedList(ruleAlarmModelListMap.get(monitorId));
            System.out.println("监控类型:"+monitorId+" value:"+ruleAlarmResultModelList);

            //对value集合按照时间排序
            ruleAlarmResultModelList.sort((o1, o2) ->{
                long terminalTime1 = DateUtil.convertStringToDate(o1.getTerminalTime()).getTime();
                long terminalTime2 = DateUtil.convertStringToDate(o2.getTerminalTime()).getTime();
                if(terminalTime1>terminalTime2){
                    return 1;
                }else if(terminalTime1<terminalTime2){
                    return -1;
                }else {
                    return 0;
                }
            });

            //todo 8遍历集合中每条数据
            int count = 0;
            for (int i = 0; i < ruleAlarmResultModelList.size(); i++) {
                //todo 8.1获取第一条数据
                CustomRuleAlarmResultModel ruleAlarmResultModel = ruleAlarmResultModelList.get(i);
                //todo 8.2获取到历史存储的count（告警出现的次数） //查询条件
                BasicDBObject queryObject = new BasicDBObject("vin", ruleAlarmResultModel.getVin())
                        .append("monitor_id", ruleAlarmResultModel.getMonitorId())
                        .append("end_time", null);
                count = getAlarmCount(collection,queryObject);
                System.out.println("告警次数："+count);

                //todo 8.3获取历史存储的告警数据
                MongoCursor<Document> historyResultDataDocument = collection.find(queryObject).iterator();
                if(historyResultDataDocument.hasNext()){
                    //找到历史的告警周期结束时间为空的数据，并根据写入数据的时间倒序排序(找到告警周期生效内的最新的一条告警数据）
                    Document lastAlarmDocument = collection.find(queryObject).sort(new BasicDBObject("process_time", -1)).first();
                    //判断当前数据是否是告警数据
                    if(ruleAlarmResultModel.getAlarmFlag()==1){
                        count+=1;
                        //如果是告警状态为未开始的时候
                        if(lastAlarmDocument.getInteger("state")==-1){
                            if(count>=ruleAlarmResultModel.getAlarmFrame()){ //告警阈值为1
                                //返回生成更新操作的document对象
                                Document docUpdate = getUpdateDocument(ruleAlarmResultModel);
                                docUpdate.append("state",1);  //将状态更新为告警中
                                docUpdate.append("count",count);
                                UpdateDocument(collection,queryObject,docUpdate);
                            }else {
                                //返回生成更新操作的document对象
                                Document docUpdate = getUpdateDocument(ruleAlarmResultModel);
                                docUpdate.append("count",count); //重置告警次数
                                UpdateDocument(collection,queryObject,docUpdate);
                            }
                        }else if(lastAlarmDocument.getInteger("state")==1){
                            //如果告警状态为告警中的状态
                            Document docUpdate = getUpdateDocument(ruleAlarmResultModel);
                            docUpdate.append("count",count); //重置告警次数
                            UpdateDocument(collection,queryObject,docUpdate);
                        }
                    }else {  //如果是非告警数据
                        //如果告警状态是未开始，则直接删除掉该状态数据即可
                        if(lastAlarmDocument.getInteger("state")==-1){
                            //删除未开始的告警数据即可
                            DeleteDocument(collection,lastAlarmDocument);
                        }else if(lastAlarmDocument.getInteger("state")==1){
                            //如果是告警状态，则将告警数据修改为告警结束，并修改告警结束时间
                            Document docUpdate = getUpdateDocument(ruleAlarmResultModel);
                            docUpdate.append("state",0); //修改为告警结束
                            //将当前数据终端时间设置为告警结束时间
                            docUpdate.append("end_time",ruleAlarmResultModel.getTerminalTime());
                            UpdateDocument(collection,queryObject,docUpdate);
                        }
                    }
                }else { //没有历史数据的场景(
                    if(ruleAlarmResultModel.getAlarmFlag()==1){
                        //需要插入一条新的告警周期的数据
                        Document docInsert = getInsertDocument(ruleAlarmResultModel);
                        docInsert.append("count",1);
                        //追加新的告警数据
                        collection.insertOne(docInsert);
                    }

                }

            }
        }

    }

    //生成插入操作的doc对象
    private Document getInsertDocument(CustomRuleAlarmResultModel ruleAlarmResultModel) {
        Document document = new Document();
        document.append("vin",ruleAlarmResultModel.getVin());
        document.append("start_time",ruleAlarmResultModel.getTerminalTime()); //告警开始时间，（这个未必是真实开始，只有状态改为告警中才是真实开始）
        if(ruleAlarmResultModel.getAlarmFrame()==1){
            document.append("end_time",ruleAlarmResultModel.getTerminalTime());//开始即结束
        }else {
            document.append("end_time",null);
        }
        document.append("rule_name",ruleAlarmResultModel.getRuleName());
        document.append("series_name",ruleAlarmResultModel.getSeriesNameValue());
        document.append("series_code",ruleAlarmResultModel.getSeriesCodeValue());
        document.append("model_name",ruleAlarmResultModel.getModelNameValue());
        document.append("model_code",ruleAlarmResultModel.getModelCodeValue());
        document.append("province",ruleAlarmResultModel.getProvince());
        document.append("city",ruleAlarmResultModel.getCity());
        document.append("monitor_id",ruleAlarmResultModel.getMonitorId());
        document.append("lat",ruleAlarmResultModel.getLat());
        document.append("lng",ruleAlarmResultModel.getLng());
        if(ruleAlarmResultModel.getAlarmFrame()<=1){
            document.append("state",0); //告警结束
        }else {
            document.append("state",-1);//未开始
        }
        document.append("process_time",String.valueOf(DateUtil.convertStringToDate(ruleAlarmResultModel.getTerminalTime()).getTime()));
        return document;
    }

    private void DeleteDocument(MongoCollection<Document> collection, Document lastAlarmDocument) {
        collection.deleteOne(lastAlarmDocument);
    }

    private Document getUpdateDocument(CustomRuleAlarmResultModel ruleAlarmResultModel) {
        Document document = new Document();
        document.append("process_time",String.valueOf(DateUtil.convertStringToDate(ruleAlarmResultModel.getTerminalTime()).getTime()));
        return document;
    }

    //更新document对象
    private void UpdateDocument(MongoCollection<Document> collection, BasicDBObject queryObject, Document docUpdate) {
        Document lastAlarmDocument = collection.find(queryObject).sort(new BasicDBObject("process_time", -1)).first();
        collection.updateOne(lastAlarmDocument,new Document("$set",docUpdate));
    }



    //获取历史的未告警结束的告警次数
    private int getAlarmCount(MongoCollection<Document> collection, BasicDBObject queryObject) {
        Document lastAlarmDocument = collection.find(queryObject)
                .sort(new BasicDBObject("process_time", -1)).first(); //最后一条数据

        //不存在为关闭的报警周期
        if(lastAlarmDocument==null){
            return 0;
        }else {
            return lastAlarmDocument.getInteger("count");
        }

    }
}
